package io.wins.vertx.database;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

import com.google.common.base.CaseFormat;
import io.vertx.core.AbstractVerticle;
import io.vertx.core.AsyncResult;
import io.vertx.core.CompositeFuture;
import io.vertx.core.Future;
import io.vertx.core.Handler;
import io.vertx.core.json.JsonArray;
import io.vertx.core.json.JsonObject;
import io.vertx.core.logging.Logger;
import io.vertx.core.logging.LoggerFactory;
import io.vertx.ext.jdbc.JDBCClient;
import io.vertx.ext.sql.SQLConnection;

public class JdbcExecutor extends AbstractVerticle {
    private final static Logger logger = LoggerFactory.getLogger(JdbcExecutor.class);
    public final static String CONSUMER = "jdbc-executor";
    private JDBCClient client;

    @Override
    public void start(final Future<Void> future) throws Exception {
        this.client = JDBCClient.create(this.vertx, DataSourceFactory.INSTANCE.getDataSource());
//        JdbcExecutor.logger.info("get datasource from factory: " + client);
        this.vertx.eventBus().consumer(JdbcExecutor.CONSUMER, message -> {
            final JsonObject jdbcMessage = (JsonObject) message.body();
            this.process(jdbcMessage, result -> {
                if (result.succeeded()) {
                    message.reply(result.result());
                }
            });
        });
        logger.info("JDBC deploy done!");
        future.complete();
    }

    private void process(final JsonObject jdbcMessage, final Handler<AsyncResult<JsonArray>> next) {
        final String action = jdbcMessage.getString("action");

        this.client.getConnection(res -> {
            if (res.failed()) {
                next.handle(Future.failedFuture(res.cause()));
                return;
            }
            final SQLConnection conn = res.result();
            final boolean needTx = Arrays.asList("insert", "update", "delete").contains(action);
            this.startConn(conn, needTx, startRes -> {
                if (startRes.failed()) {
                    next.handle(Future.failedFuture(startRes.cause()));
                    conn.close();
                    return;
                }

                switch (action) {
                    case "insert":
                    case "delete":
                    case "update":
                        this.doUpdate(conn, jdbcMessage, result -> {
                            if (result.failed()) {
                                next.handle(Future.failedFuture(result.cause()));
                                conn.close();
                                return;
                            }

                            this.endConn(conn, needTx, endRes -> {
                                next.handle(Future.succeededFuture(result.result()));
                            });
                        });
                        break;
                    case "select":
                        this.doQuery(conn, jdbcMessage, result -> {
                            if (result.failed()) {
                                next.handle(Future.failedFuture(result.cause()));
                                conn.close();
                                return;
                            }

                            this.endConn(conn, needTx, endRes -> {
                                next.handle(Future.succeededFuture(result.result()));
                            });
                        });
                        break;
                    case "batch":
                        final List<JsonObject> actions = jdbcMessage.getJsonArray("subActions").getList();

                        final List<Future> futures = new ArrayList<Future>();
                        actions.forEach(actMsg -> {
                            final Future<JsonArray> fut = Future.future();
                            futures.add(fut);
                            this.doUpdate(conn, actMsg, fut.completer());
                        });

                        CompositeFuture.all(futures).setHandler(batchRes -> {
                            if (batchRes.failed()) {
                                next.handle(Future.failedFuture(batchRes.cause()));
                                conn.close();
                                return;
                            }

                            this.endConn(conn, needTx, endRes -> {
                                next.handle(Future.succeededFuture(new JsonArray(batchRes.result().list())));
                            });
                        });
                }
            });
        });
    }

    protected void doUpdate(final SQLConnection conn, final JsonObject jsonMessage,
                            final Handler<AsyncResult<JsonArray>> next) {
        final String sql = jsonMessage.getString("sql");
        final JsonArray params = jsonMessage.getJsonArray("params");

        List<JsonArray> batchParams = new ArrayList<>();
        if (params != null) {
            if (params.size() > 0 && params.getValue(0) instanceof JsonArray) {
                batchParams = params.getList();
            } else {
                batchParams.add(params);
            }
        }

        conn.batchWithParams(sql, batchParams, res -> {
            if (res.failed()) {
                next.handle(Future.failedFuture(res.cause()));
            }
            next.handle(Future.succeededFuture(new JsonArray(res.result())));
        });
    }

    private void doQuery(final SQLConnection conn, final JsonObject queryMessage,
                         final Handler<AsyncResult<JsonArray>> next) {
        final String sql = queryMessage.getString("sql");
        final JsonArray params = queryMessage.getJsonArray("params");
        conn.queryWithParams(sql, params, res -> {
            if (res.failed()) {
                logger.error("Query Error by: {}", res.cause().getMessage());
                next.handle(Future.failedFuture(res.cause()));
            }
            this.toCamelCase(res.result().getColumnNames());
            next.handle(Future.succeededFuture(new JsonArray(res.result().getRows())));
        });
    }

    private void startConn(final SQLConnection conn, final boolean startTx, final Handler<AsyncResult<Void>> done) {
        if (startTx) {
            conn.setAutoCommit(false, res -> {
                if (res.failed()) {
                    done.handle(Future.failedFuture(res.cause()));
                    return;
                }
                done.handle(Future.succeededFuture());
            });
        } else {
            done.handle(Future.succeededFuture());
        }
    }

    private void endConn(final SQLConnection conn, final boolean endTx, final Handler<AsyncResult<Void>> done) {
        if (endTx) {
            conn.commit(res -> {
                if (res.failed()) {
                    done.handle(Future.failedFuture(res.cause()));
                    return;
                }
                this.closeConn(conn, done);
            });
        } else {
            this.closeConn(conn, done);
        }
    }

    private void closeConn(final SQLConnection conn, final Handler<AsyncResult<Void>> done) {
        conn.close(res -> {
            if (res.failed()) {
                done.handle(Future.failedFuture(res.cause()));
                return;
            }
            done.handle(Future.succeededFuture());
        });
    }

    private void toCamelCase(final List<String> list) {
        if (list != null && list.size() > 0) {
            int i = 0;
            while (i < list.size()) {
                final String s = list.get(i);
                if (s == null || s.indexOf('_') < 0) {
                    i++;
                    continue;
                }
                list.set(i, CaseFormat.LOWER_UNDERSCORE.to(CaseFormat.LOWER_CAMEL, list.get(i)));
            }
        }
    }

}
